"""
异步管理器模块
简化复杂的异步/同步事件循环处理逻辑，提供统一的异步操作接口

作者: 架构重构优化
日期: 2025-01-05
"""

import asyncio
import logging
import threading
from typing import Any, Callable, Coroutine, Optional, TypeVar, Union
from functools import wraps
import concurrent.futures

logger = logging.getLogger(__name__)

T = TypeVar('T')

class AsyncManager:
    """
    异步管理器
    提供统一的异步/同步操作接口，简化事件循环处理

    性能优化版本：
    - 支持长期运行的后台事件循环
    - 更好的资源管理
    - 减少事件循环创建/销毁开销
    """

    def __init__(self):
        """初始化异步管理器"""
        self._thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=4)
        self._loop_cache = {}
        self._lock = threading.Lock()

        # 长期运行的后台事件循环
        self._background_loop = None
        self._background_thread = None
        self._background_loop_ready = threading.Event()

        logger.info("异步管理器初始化完成")

    def _ensure_background_loop(self):
        """确保后台事件循环运行"""
        if self._background_loop is None or self._background_loop.is_closed():
            with self._lock:
                if self._background_loop is None or self._background_loop.is_closed():
                    self._start_background_loop()

    def _start_background_loop(self):
        """启动后台事件循环"""
        def run_background_loop():
            """后台线程中运行事件循环"""
            self._background_loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self._background_loop)
            self._background_loop_ready.set()

            try:
                self._background_loop.run_forever()
            except Exception as e:
                logger.error(f"后台事件循环异常: {e}")
            finally:
                self._background_loop.close()

        self._background_thread = threading.Thread(target=run_background_loop, daemon=True)
        self._background_thread.start()

        # 等待后台循环启动
        self._background_loop_ready.wait(timeout=5.0)
        if not self._background_loop_ready.is_set():
            raise RuntimeError("后台事件循环启动超时")

        logger.info("后台事件循环启动成功")

    def run_async_safe(self, coro: Coroutine[Any, Any, T], timeout: float = 30.0) -> Optional[T]:
        """
        安全地运行异步协程，自动处理事件循环冲突

        注意：在异步存储移除后，大部分异步操作已不再需要
        此方法主要用于向后兼容和特殊情况

        Args:
            coro: 要运行的协程
            timeout: 超时时间（秒）

        Returns:
            协程的返回值，如果失败则返回 None
        """
        try:
            # 检查是否真的需要异步执行
            logger.debug(f"异步操作请求: {type(coro).__name__}")

            # 方法1：检查当前线程是否有运行中的事件循环
            try:
                current_loop = asyncio.get_running_loop()
                if not current_loop.is_closed():
                    # 在当前循环中使用run_coroutine_threadsafe
                    logger.debug("在当前事件循环中使用线程安全执行")
                    future = asyncio.run_coroutine_threadsafe(coro, current_loop)
                    return future.result(timeout)
            except RuntimeError:
                # 没有运行中的事件循环
                pass

            # 方法2：使用后台事件循环（仅在必要时）
            try:
                self._ensure_background_loop()
                if self._background_loop and not self._background_loop.is_closed():
                    logger.debug("使用后台事件循环执行")
                    future = asyncio.run_coroutine_threadsafe(coro, self._background_loop)
                    return future.result(timeout)
            except Exception as e:
                logger.debug(f"后台事件循环执行失败: {e}")

            # 方法3：创建新的事件循环（最后的选择）
            logger.debug("创建新的临时事件循环")
            return self._run_with_new_loop(coro, timeout)

        except asyncio.TimeoutError:
            logger.debug(f"异步操作超时: {timeout}秒 - 这在异步存储移除后是正常的")
            return None
        except Exception as e:
            logger.debug(f"异步操作失败: {e} - 可能是由于异步存储移除")
            return None

    def _run_in_thread_pool(self, coro: Coroutine[Any, Any, T], timeout: float) -> Optional[T]:
        """在线程池中运行协程"""
        try:
            def run_in_new_loop():
                # 在新线程中创建新的事件循环
                new_loop = asyncio.new_event_loop()
                asyncio.set_event_loop(new_loop)
                try:
                    return new_loop.run_until_complete(asyncio.wait_for(coro, timeout))
                finally:
                    new_loop.close()
                    asyncio.set_event_loop(None)
            
            # 使用线程池执行
            future = self._thread_pool.submit(run_in_new_loop)
            return future.result(timeout=timeout + 5)  # 额外5秒缓冲
            
        except concurrent.futures.TimeoutError:
            logger.warning("线程池执行超时")
            return None
        except Exception as e:
            logger.error(f"线程池执行失败: {e}")
            return None

    def _run_with_new_loop(self, coro: Coroutine[Any, Any, T], timeout: float) -> Optional[T]:
        """使用新的事件循环运行协程，确保所有操作完成"""
        loop = None
        try:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)

            # 运行协程并等待完成
            result = loop.run_until_complete(asyncio.wait_for(coro, timeout))

            # 确保所有待处理的任务完成
            pending_tasks = asyncio.all_tasks(loop)
            if pending_tasks:
                logger.debug(f"等待 {len(pending_tasks)} 个待处理任务完成")
                loop.run_until_complete(asyncio.gather(*pending_tasks, return_exceptions=True))

            return result

        finally:
            if loop:
                try:
                    # 再次检查是否有新的待处理任务
                    remaining_tasks = asyncio.all_tasks(loop)
                    if remaining_tasks:
                        logger.debug(f"清理 {len(remaining_tasks)} 个剩余任务")
                        loop.run_until_complete(asyncio.gather(*remaining_tasks, return_exceptions=True))

                    # 安全关闭事件循环
                    if not loop.is_closed():
                        loop.close()

                except Exception as e:
                    logger.warning(f"关闭事件循环失败: {e}")
                try:
                    asyncio.set_event_loop(None)
                except Exception as e:
                    logger.warning(f"重置事件循环失败: {e}")

    def create_task_safe(self, coro: Coroutine[Any, Any, T]) -> Optional[asyncio.Task]:
        """
        安全地创建异步任务

        Args:
            coro: 要创建任务的协程

        Returns:
            创建的任务，如果失败则返回 None
        """
        try:
            # 尝试在当前事件循环中创建任务
            try:
                loop = asyncio.get_running_loop()
                if not loop.is_closed():
                    return loop.create_task(coro)
            except RuntimeError:
                pass

            # 使用后台事件循环创建任务
            self._ensure_background_loop()
            if self._background_loop and not self._background_loop.is_closed():
                # 在后台循环中创建任务
                future = asyncio.run_coroutine_threadsafe(coro, self._background_loop)
                return future  # 返回Future对象，可以当作Task使用

            logger.warning("无法创建异步任务，所有事件循环都不可用")
            return None

        except Exception as e:
            logger.error(f"创建任务失败: {e}")
            return None

    def run_sync_in_async(self, func: Callable[..., T], *args, **kwargs) -> Coroutine[Any, Any, T]:
        """
        在异步环境中运行同步函数
        
        Args:
            func: 要运行的同步函数
            *args: 函数参数
            **kwargs: 函数关键字参数
            
        Returns:
            包装后的协程
        """
        async def wrapper():
            try:
                loop = asyncio.get_running_loop()
                return await loop.run_in_executor(self._thread_pool, func, *args, **kwargs)
            except Exception as e:
                logger.error(f"同步函数异步执行失败: {e}")
                raise
        
        return wrapper()

    def is_async_context(self) -> bool:
        """
        检查当前是否在异步上下文中
        
        Returns:
            是否在异步上下文中
        """
        try:
            asyncio.get_running_loop()
            return True
        except RuntimeError:
            return False

    def close(self):
        """关闭异步管理器，清理资源"""
        try:
            # 关闭后台事件循环
            if self._background_loop and not self._background_loop.is_closed():
                self._background_loop.call_soon_threadsafe(self._background_loop.stop)
                if self._background_thread and self._background_thread.is_alive():
                    self._background_thread.join(timeout=5.0)

            # 关闭线程池
            self._thread_pool.shutdown(wait=True)
            logger.info("异步管理器已关闭")
        except Exception as e:
            logger.error(f"关闭异步管理器失败: {e}")

    def __del__(self):
        """析构函数，确保资源清理"""
        try:
            self.close()
        except:
            pass

# 全局异步管理器实例
_async_manager_instance = None
_async_manager_lock = threading.Lock()

def get_async_manager() -> AsyncManager:
    """获取全局异步管理器实例（单例模式）"""
    global _async_manager_instance
    if _async_manager_instance is None:
        with _async_manager_lock:
            if _async_manager_instance is None:
                _async_manager_instance = AsyncManager()
    return _async_manager_instance

def reset_async_manager():
    """重置全局异步管理器实例（用于测试）"""
    global _async_manager_instance
    with _async_manager_lock:
        if _async_manager_instance:
            _async_manager_instance.close()
        _async_manager_instance = None

# 便捷装饰器
def async_safe(timeout: float = 30.0):
    """
    装饰器：使异步函数可以安全地在任何环境中调用
    
    Args:
        timeout: 超时时间（秒）
    """
    def decorator(func: Callable[..., Coroutine[Any, Any, T]]) -> Callable[..., Optional[T]]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Optional[T]:
            coro = func(*args, **kwargs)
            manager = get_async_manager()
            return manager.run_async_safe(coro, timeout)
        return wrapper
    return decorator

def sync_to_async(func: Callable[..., T]) -> Callable[..., Coroutine[Any, Any, T]]:
    """
    装饰器：将同步函数转换为异步函数
    
    Args:
        func: 要转换的同步函数
    """
    @wraps(func)
    async def wrapper(*args, **kwargs) -> T:
        manager = get_async_manager()
        return await manager.run_sync_in_async(func, *args, **kwargs)
    return wrapper

# 便捷函数
def run_async(coro: Coroutine[Any, Any, T], timeout: float = 30.0) -> Optional[T]:
    """
    便捷函数：安全地运行异步协程
    
    Args:
        coro: 要运行的协程
        timeout: 超时时间（秒）
        
    Returns:
        协程的返回值
    """
    manager = get_async_manager()
    return manager.run_async_safe(coro, timeout)

def create_task(coro: Coroutine[Any, Any, T]) -> Optional[asyncio.Task]:
    """
    便捷函数：安全地创建异步任务
    
    Args:
        coro: 要创建任务的协程
        
    Returns:
        创建的任务
    """
    manager = get_async_manager()
    return manager.create_task_safe(coro)
